package com.rabbit.chat.core;

import java.io.IOException;
import java.sql.Timestamp;
import java.util.ArrayList;

import com.rabbitmq.client.AMQP.BasicProperties;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.ConnectionFactory;
import com.rabbitmq.client.ConsumerCancelledException;
import com.rabbitmq.client.QueueingConsumer;
import com.rabbitmq.client.ShutdownSignalException;

public class RabbitChat implements IRabbitChat, Runnable {

	private String serverUsername = "chat";
	private String queueHost = "192.168.0.100";
	private String virtualHost = "ConsoleChatTest";
	private String exchangeName = "ChatConsole";
	private String clientBaseQueue = "chat-console-queue-";

	private static final String DEFAULT_EXCHANGE_NAME = "ChatConsole";
	
	private Connection connection;
	private Channel channel;
	private String replyQueueName;
	private QueueingConsumer consumer;
	private String routingKey;
	private Thread runner;

	public RabbitChat(String clientName) {
		this(clientName, DEFAULT_EXCHANGE_NAME);
	}

	public RabbitChat(String clientName, String room) { 
		this.routingKey = clientName;
		this.exchangeName = room;
	}

	public IRabbitChat Initialize() throws IOException{
		ConnectionFactory factory = new ConnectionFactory();
		factory.setHost(queueHost);
		factory.setUsername(serverUsername);
		factory.setPassword(serverUsername);
		factory.setVirtualHost(virtualHost);

		connection = factory.newConnection();
		channel = connection.createChannel();

		channel.exchangeDeclare(exchangeName,"topic",true);

		this.replyQueueName = clientBaseQueue + routingKey; 
		channel.queueDeclare(replyQueueName, true,false,false,null);
		channel.queueBind(replyQueueName, exchangeName,"*");

		channel.basicQos(1);

		consumer = new QueueingConsumer(channel);

		channel.basicConsume(replyQueueName, false, consumer);

		this.runner = new Thread(this);
		this.runner.start();
		return this;
	}

	private void ConsumeMessage() throws ShutdownSignalException, ConsumerCancelledException, InterruptedException, IOException  {
		while (true) {
			QueueingConsumer.Delivery delivery = consumer.nextDelivery();

			String message = new String(delivery.getBody());
			String routingKey = delivery.getEnvelope().getRoutingKey();

			channel.basicAck(delivery.getEnvelope().getDeliveryTag(), false);
			if(routingKey.equals(this.routingKey))
			{	
				OnLastMessageSent();
			}
			OnMessageReceive(routingKey, message);
			
		}
	}


	
	private ArrayList<String> coorIdList = new ArrayList<String>();

	public void SendMessage(String message) throws Exception {
		String corrId = java.util.UUID.randomUUID().toString();
		coorIdList.add(corrId);

		BasicProperties props = new BasicProperties
				.Builder()
		.correlationId(corrId)
		.replyTo(replyQueueName)
		.timestamp(new Timestamp(System.currentTimeMillis()))
		.build();

		channel.basicPublish(exchangeName, routingKey, props, message.getBytes());
	}

	public void run() {
		try {
			ConsumeMessage();
		} catch (Exception e) {
			OnError(e);
		}
	}
	
	public void OnMessageReceive(String sender, String message) {
		// TODO Auto-generated method stub

	}

	public void OnLastMessageSent() {
		// TODO Auto-generated method stub

	}

	public void OnError(Exception e){

	}

	public void close() throws Exception {

		connection.close();
	}
}
